{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Import pixiedust\n",
    "Start by importing pixiedust which if all bootstrap and install steps were run correctly.\n",
    "You should see below for opening the pixiedust database successfully with no errors.\n",
    "Depending on the version of pixiedust that gets installed, it may ask you to update.\n",
    "If so, run this first cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!pip install --user --upgrade pixiedust"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pixiedust\n",
    "import geowave_pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pixiedust.enableJobMonitor()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating the SQLContext and inspecting pyspark Context\n",
    "Pixiedust imports pyspark and the SparkContext + SparkSession should be already available through the \"sc\" and \"spark\" variables respectively."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print Spark info and create sql_context\n",
    "print('Spark Version: {0}'.format(sc.version))\n",
    "print('Python Version: {0}'.format(sc.pythonVer))\n",
    "print('Application Name: {0}'.format(sc.appName))\n",
    "print('Application ID: {0}'.format(sc.applicationId))\n",
    "print('Spark Master: {0}'.format( sc.master))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Download GDELT Data\n",
    "Download the data necessary to perform Kmeans"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "cd /mnt/tmp\n",
    "wget s3.amazonaws.com/geowave/latest/scripts/emr/quickstart/geowave-env.sh\n",
    "source /mnt/tmp/geowave-env.sh\n",
    "mkdir gdelt\n",
    "cd gdelt\n",
    "wget http://data.gdeltproject.org/events/md5sums\n",
    "for file in `cat md5sums | cut -d' ' -f3 | grep \"^${TIME_REGEX}\"` ; \\\n",
    "do wget http://data.gdeltproject.org/events/$file ; done\n",
    "md5sum -c md5sums 2>&1 | grep \"^${TIME_REGEX}\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create datastores and ingest gdelt data.\n",
    "The ingest process may take a few minutes. If the '*' is present left of the cell the command is still running. Output will not appear below under the process is finished."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "\n",
    "# We have to source here again because bash runs in a separate sub process each cell.\n",
    "source /mnt/tmp/geowave-env.sh\n",
    "\n",
    "# clear old potential runs\n",
    "geowave store clear gdelt\n",
    "geowave store rm gdelt\n",
    "geowave store clear kmeans_gdelt\n",
    "geowave store rm kmeans_gdelt\n",
    "\n",
    "# configure geowave connection params for hbase stores \"gdelt\" and \"kmeans\"\n",
    "geowave store add gdelt --gwNamespace geowave.gdelt -t hbase --zookeeper $HOSTNAME:2181\n",
    "geowave store add kmeans_gdelt --gwNamespace geowave.kmeans -t hbase --zookeeper $HOSTNAME:2181\n",
    "\n",
    "# configure a spatial index\n",
    "geowave index add gelt gdeltspatial -t spatial --partitionStrategy round_robin --numPartitions $NUM_PARTITIONS\n",
    "\n",
    "# run the ingest for a 10x10 deg bounding box over Europe\n",
    "geowave ingest localtogw /mnt/tmp/gdelt gdelt gdeltspatial -f gdelt \\\n",
    "--gdelt.cql \"BBOX(geometry, 0, 50, 10, 60)\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Run KMeans\n",
    "Running the KMeans process may take a few minutes you should be able to track the progress of the task via the console or Spark History Server once the job begins."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "# clear out potential old runs\n",
    "geowave store clear kmeans_gdelt",
    "\n",
    "# configure a spatial index\n",
    "geowave index add kmeans_gdelt gdeltspatial -t spatial --partitionStrategy round_robin --numPartitions $NUM_PARTITIONS\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#grab classes from jvm\n",
    "\n",
    "# Pull classes to desribe core GeoWave classes\n",
    "hbase_options_class = sc._jvm.org.locationtech.geowave.datastore.hbase.cli.config.HBaseRequiredOptions\n",
    "query_options_class = sc._jvm.org.locationtech.geowave.core.store.query.QueryOptions\n",
    "byte_array_class = sc._jvm.org.locationtech.geowave.core.index.ByteArrayId\n",
    "# Pull core GeoWave Spark classes from jvm\n",
    "geowave_rdd_class = sc._jvm.org.locationtech.geowave.analytic.spark.GeoWaveRDD\n",
    "rdd_loader_class = sc._jvm.org.locationtech.geowave.analytic.spark.GeoWaveRDDLoader\n",
    "rdd_options_class = sc._jvm.org.locationtech.geowave.analytic.spark.RDDOptions\n",
    "sf_df_class = sc._jvm.org.locationtech.geowave.analytic.spark.sparksql.SimpleFeatureDataFrame\n",
    "kmeans_runner_class = sc._jvm.org.locationtech.geowave.analytic.spark.kmeans.KMeansRunner\n",
    "\n",
    "datastore_utils_class = sc._jvm.org.locationtech.geowave.core.store.util.DataStoreUtils\n",
    "\n",
    "spatial_encoders_class = sc._jvm.org.locationtech.geowave.analytic.spark.sparksql.GeoWaveSpatialEncoders\n",
    "\n",
    "spatial_encoders_class.registerUDTs()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Setup input datastore options\n",
    "input_store = hbase_options_class()\n",
    "input_store.setZookeeper(os.environ['HOSTNAME'] + ':2181')\n",
    "input_store.setGeowaveNamespace('geowave.gdelt')\n",
    "\n",
    "#Setup output datastore options\n",
    "output_store = hbase_options_class()\n",
    "output_store.setZookeeper(os.environ['HOSTNAME'] + ':2181')\n",
    "output_store.setGeowaveNamespace('geowave.kmeans')\n",
    "\n",
    "#Create a instance of the runner, and datastore options\n",
    "kmeans_runner = kmeans_runner_class()\n",
    "input_store_plugin = input_store.createPluginOptions()\n",
    "output_store_plugin = output_store.createPluginOptions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Set the appropriate properties\n",
    "kmeans_runner.setSparkSession(sc._jsparkSession)\n",
    "\n",
    "kmeans_runner.setAdapterId('gdeltevent')\n",
    "kmeans_runner.setInputDataStore(input_store_plugin)\n",
    "kmeans_runner.setOutputDataStore(output_store_plugin)\n",
    "kmeans_runner.setCqlFilter(\"BBOX(geometry, 0, 50, 10, 60)\")\n",
    "kmeans_runner.setCentroidTypeName('mycentroids_gdelt')\n",
    "kmeans_runner.setHullTypeName('myhulls_gdelt')\n",
    "kmeans_runner.setGenerateHulls(True)\n",
    "kmeans_runner.setComputeHullData(True)\n",
    "#Execute the kmeans runner\n",
    "kmeans_runner.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load resulting Centroids into DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the dataframe and get a rdd for the output of kmeans\n",
    "adapter_id = byte_array_class('mycentroids_gdelt')\n",
    "query_adapter = datastore_utils_class.getDataAdapter(output_store_plugin, adapter_id)\n",
    "query_options = query_options_class(query_adapter)\n",
    "\n",
    "# Create RDDOptions for loader\n",
    "rdd_options = rdd_options_class()\n",
    "rdd_options.setQueryOptions(query_options)\n",
    "output_rdd = rdd_loader_class.loadRDD(sc._jsc.sc(), output_store_plugin, rdd_options)\n",
    "\n",
    "# Create a SimpleFeatureDataFrame from the GeoWaveRDD\n",
    "sf_df = sf_df_class(spark._jsparkSession)\n",
    "sf_df.init(output_store_plugin, adapter_id)\n",
    "df = sf_df.getDataFrame(output_rdd)\n",
    "\n",
    "# Convert Java DataFrame to Python DataFrame\n",
    "import pyspark.mllib.common as convert\n",
    "py_df = convert._java2py(sc, df)\n",
    "\n",
    "py_df.createOrReplaceTempView('mycentroids')\n",
    "\n",
    "df = spark.sql(\"select * from mycentroids\")\n",
    "\n",
    "display(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Parse DataFrame data into lat/lon columns and display centroids on map\n",
    "Using pixiedust's built in map visualization we can display data on a map assuming it has the following properties.\n",
    "- Keys: put your latitude and longitude fields here. They must be floating values. These fields must be named latitude, lat or y and longitude, lon or x.\n",
    "- Values: the field you want to use to thematically color the map. Only one field can be used.\n",
    "\n",
    "Also you will need a access token from whichever map renderer you choose to use with pixiedust (mapbox, google).\n",
    "Follow the instructions in the token help on how to create and use the access token."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert the string point information into lat long columns and create a new dataframe for those.\n",
    "import pyspark\n",
    "def parseRow(row):\n",
    "    lat=row.geom.y\n",
    "    lon=row.geom.x\n",
    "    return pyspark.sql.Row(lat=lat,lon=lon,ClusterIndex=row.ClusterIndex)\n",
    "    \n",
    "row_rdd = df.rdd\n",
    "\n",
    "new_rdd = row_rdd.map(lambda row: parseRow(row))\n",
    "\n",
    "new_df = new_rdd.toDF()\n",
    "display(new_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Export KMeans Hulls to DataFrame\n",
    "If you have some more complex data to visualize pixiedust may not be the best option.\n",
    "\n",
    "The Kmeans hull generation outputs polygons that would be difficult for pixiedust to display without\n",
    "creating a special plugin. \n",
    "\n",
    "Instead, we can use another map renderer to visualize our data. For the Kmeans hulls we will use folium to visualize the data. Folium allows us to easily add wms layers to our notebook, and we can combine that with GeoWaves geoserver functionality to render the hulls and centroids. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the dataframe and get a rdd for the output of kmeans\n",
    "# Grab adapter and setup query options for rdd load\n",
    "adapter_id = byte_array_class('myhulls_gdelt')\n",
    "query_adapter = datastore_utils_class.getDataAdapter(output_store_plugin, adapter_id)\n",
    "query_options = query_options_class(query_adapter)\n",
    "\n",
    "# Use GeoWaveRDDLoader to load an RDD\n",
    "rdd_options = rdd_options_class()\n",
    "rdd_options.setQueryOptions(query_options)\n",
    "output_rdd_hulls = rdd_loader_class.loadRDD(sc._jsc.sc(), output_store_plugin, rdd_options)\n",
    "\n",
    "# Create a SimpleFeatureDataFrame from the GeoWaveRDD\n",
    "sf_df_hulls = sf_df_class(spark._jsparkSession)\n",
    "sf_df_hulls.init(output_store_plugin, adapter_id)\n",
    "df_hulls = sf_df_hulls.getDataFrame(output_rdd_hulls)\n",
    "\n",
    "# Convert Java DataFrame to Python DataFrame\n",
    "import pyspark.mllib.common as convert\n",
    "py_df_hulls = convert._java2py(sc, df_hulls)\n",
    "\n",
    "# Create a sql table view of the hulls data\n",
    "py_df_hulls.createOrReplaceTempView('myhulls')\n",
    "\n",
    "# Run SQL Query on Hulls data\n",
    "df_hulls = spark.sql(\"select * from myhulls order by Density\")\n",
    "\n",
    "display(df_hulls)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Visualize results using geoserver and wms\n",
    "folium provides an easy way to visualize leaflet maps in jupyter notebooks. \n",
    "When the data is too complicated or big to work within the simple framework pixiedust provides for map display we can instead turn to geoserver and wms to render our layers. First we configure geoserver then setup wms layers for folium to display the kmeans results on the map."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "# set up geoserver\n",
    "geowave config geoserver \"$HOSTNAME:8000\"\n",
    "\n",
    "# add the centroids layer\n",
    "geowave gs layer add kmeans_gdelt -id mycentroids_gdelt\n",
    "geowave gs style set mycentroids_gdelt --styleName point\n",
    "\n",
    "# add the hulls layer\n",
    "geowave gs layer add kmeans_gdelt -id myhulls_gdelt\n",
    "geowave gs style set myhulls_gdelt --styleName line"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import owslib\n",
    "from owslib.wms import WebMapService\n",
    "\n",
    "url = \"http://\" + os.environ['HOSTNAME'] + \":8000/geoserver/geowave/wms\"\n",
    "web_map_services = WebMapService(url)\n",
    "\n",
    "#print layers available wms\n",
    "print('\\n'.join(web_map_services.contents.keys()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import folium\n",
    "\n",
    "#grab wms info for centroids\n",
    "layer = 'mycentroids_gdelt'\n",
    "wms = web_map_services.contents[layer]\n",
    "\n",
    "#build center of map off centroid bbox\n",
    "lon = (wms.boundingBox[0] + wms.boundingBox[2]) / 2.\n",
    "lat = (wms.boundingBox[1] + wms.boundingBox[3]) / 2.\n",
    "center = [lat, lon]\n",
    "\n",
    "m = folium.Map(location = center,zoom_start=3)\n",
    "\n",
    "\n",
    "name = wms.title\n",
    "centroids = folium.raster_layers.WmsTileLayer(\n",
    "    url=url,\n",
    "    name=name,\n",
    "    fmt='image/png',\n",
    "    transparent=True,\n",
    "    layers=layer,\n",
    "    overlay=True,\n",
    "    COLORSCALERANGE='1.2,28',\n",
    ")\n",
    "centroids.add_to(m)\n",
    "\n",
    "layer = 'myhulls_gdelt'\n",
    "wms = web_map_services.contents[layer]\n",
    "\n",
    "name = wms.title\n",
    "hulls = folium.raster_layers.WmsTileLayer(\n",
    "    url=url,\n",
    "    name=name,\n",
    "    fmt='image/png',\n",
    "    transparent=True,\n",
    "    layers=layer,\n",
    "    overlay=True,\n",
    "    COLORSCALERANGE='1.2,28',\n",
    ")\n",
    "hulls.add_to(m)\n",
    "m"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    ""
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python with Pixiedust (Spark 2.3)",
   "language": "python",
   "name": "pythonwithpixiedustspark23"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3.0
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}